package com.zp.datax.core;

import com.zp.datax.core.function.BiFunctionWrapper;
import com.zp.datax.core.function.ConsumerWrapper;
import com.zp.datax.core.model.DataxJobLog;
import com.zp.datax.core.result.ReturnT;
import com.zp.datax.core.util.ThreadPoolFactory;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.Reader;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 页面调用处理器
 *
 * @author bruce
 * @date 2019-07-15
 */
@Slf4j
public class PageExecProcessor extends ExecProcessor {

    public PageExecProcessor(String pathname, String strCondition) {
        super(pathname, strCondition);
    }

    public PageExecProcessor(ExecProcessor processor) {
        super(processor);
    }

    @Override
    public ReturnT exec(Handler handler, ConsumerWrapper<DataxJobLog, Exception> consumerWrapper, DataxJobLog dataxJobLog) {

        BiFunctionWrapper<Reader, Process, ReturnT, Exception> function = (r, p) -> {
            BufferedReader reader = (BufferedReader) r;
            Map<String, Object> map = new HashMap<>();
            map.put("startTime", Instant.now().toEpochMilli());

            //记录日志
            asSynSaveJobLog(consumerWrapper, dataxJobLog, map);

            String line;
            while ((line = reader.readLine()) != null) {


                log.info(line);

                Object[] objarray = handler.handler(line);
                if (!Arrays.asList(objarray).isEmpty()) {
                    map.put(Objects.toString(objarray[0]), Objects.toString(objarray[1]));
                }
            }

            p.waitFor();

            map.put("endTime", Objects.toString(Instant.now().toEpochMilli()));

            ReturnT returnT = new ReturnT();
            returnT.setData(map);

            if (isSuccess(map)) {
                returnT.setCode(ReturnT.SUCCESS_CODE);
                dataxJobLog.setMsg("同步成功");
            } else {
                returnT.setCode(ReturnT.FAIL_CODE);
                dataxJobLog.setMsg("同步失败");
            }

            return returnT;
        };

        String[] envp = new String[]{String.format("java_home=%s", System.getProperty("java.home"))};

        return exec(function, getCommandArray(), envp);
    }

    /**
     * 检验抽取数据结果
     *
     * @param map {@link Map}
     * @return true-成功；false-失败；
     */
    private boolean isSuccess(Map<String, Object> map) {

        boolean flag = false;

        if (map.containsKey("failCount")
                && map.containsKey("record")) {

            Integer failCount = Integer.valueOf(Objects.toString(map.get("failCount")));
            Integer record = Integer.valueOf(Objects.toString(map.get("record")));

            flag = record.compareTo(failCount) >= 0 ? true : false;

        } else if (map.containsKey("failCount")
                && map.containsKey("percentage")
                && map.containsKey("total")) {

            BigDecimal failCount = new BigDecimal(Objects.toString(map.get("failCount")));
            BigDecimal total = new BigDecimal(Objects.toString(map.get("total")));
            BigDecimal percentage = new BigDecimal(Objects.toString(map.get("percentage")));

            BigDecimal permissionMaxFailCount = total.multiply(percentage);
            flag = permissionMaxFailCount.compareTo(failCount) >= 0 ? true : false;

        }

        return flag;
    }

    /**
     * 异步保存日志记录
     *
     * @param consumerWrapper
     * @param dataxJobLog
     * @param map
     */
    private void asSynSaveJobLog(ConsumerWrapper<DataxJobLog, Exception> consumerWrapper, DataxJobLog dataxJobLog, Map<String, Object> map) {

        ThreadPoolFactory.INSTANCE.getThreadPool().execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(10L);
                //记录日志
                if (Objects.nonNull(dataxJobLog)) {
                    log.info("正在记录job日志信息，taskId :: {}", dataxJobLog.getTaskId());
                    dataxJobLog.setPath(getLastLogPath());
                    dataxJobLog.setName(getLogname());
                    consumerWrapper.accept(dataxJobLog);

                    map.put("dataxJobLog", dataxJobLog);
                }
            } catch (Exception e) {
                log.error("记录日志出错 :: {}", e.getMessage());
            }
        });
    }

}
